//-------------------------------------------------------------------
//	Copyright (c) 2012-2014 Zhirnov Andrey
//	This file is part of the "UXGEN-ES" project.
//	See copyright notice in "Engine.h".
//-------------------------------------------------------------------

#pragma once

#include "Base/Include/BaseObject.h"
#include "Base/Platform/Platform.h"
#include "Base/Parallel/TMultiThreadQueue.h"

namespace Engine
{
namespace Base
{
	
	//
	// Parallel Operation interface
	//
	
	template <usize BufSize = UX_OPT_FUNCTION_INTERFACE_SIZE>
	struct TParallelOp
	{
	// types
	public:
		typedef TParallelOp< BufSize >		Self;
		typedef function< void, BufSize >	func_t;


	// variables
	public:
		func_t		func;


	// methods
	public:
		TParallelOp ()
		{}

		TParallelOp (const Self &other) : func( other.func )
		{}

		RVALUE_REFERENCE_ONLY(
			TParallelOp (Self &&other) : func( ToRValueRef( other.func ) )
			{}
		)

		void Process () const
		{
			return func.Call();
		}

		void operator = (Self &other)
		{
			func.MoveTo( other.func );
		}
	};

	typedef TParallelOp<>	ParallelOp;


	SHARED_POINTER( IParallelThread );



	//
	// Parallel Base Object
	//

	class IParallelThread : public BaseObject
	{
	// types
	public:
		typedef IParallelThread			Self;


	// variables
	protected:
		EngineSubSystems		_subSystems;
		set< BaseObjectPtr >	_references;		// array of attached referenced objects


	// methods
	protected:
		IParallelThread () :
			BaseObject(EngineSubSystemsRef(_subSystems))
		{
			ESS()->SetCurrentThread( this );
		}

		~IParallelThread ()
		{
			ESS()->SetCurrentThread( null );
		}

		void _OnStart (const EngineSubSystemsRef ess)
		{
			_subSystems = *ess.raw_ptr();
			_subSystems.SetCurrentThread( this );
		}

		void _OnExit ()
		{
			_references.Clear();
		}


	public:
		void AttachObject (BaseObject *obj) const
		{
			ASSERT( IsCurrent() );
			obj->GetSubSystems().Override( ESS() );
		}

		bool AttachReferenced (const BaseObjectPtr &obj)
		{
			CHECK_ERR( IsCurrent() );
			
			_references.AddOrReplace( obj );
			return true;
		}

		bool RemoveReferenced (const BaseObjectPtr &obj)
		{
			CHECK_ERR( IsCurrent() );

			return _references.Erase( obj );
		}
		
		void EnqueueFlushCommand ()
		{
			return EnqueueFlushCommand( this );
		}

		void EnqueueFlushCommand (const IParallelThreadPtr &forThread)
		{
			ParallelOp	op;
			FunctionBuilder::Create( op.func, forThread, &IParallelThread::Flush );
			Push( ToRValueRef( op ) );
		}


	// interface
	public:
		virtual bool IsCurrent () const = 0;

		virtual void Flush () = 0;

		virtual void Push (const ParallelOp &) = 0;

		RVALUE_REFERENCE_ONLY( virtual void Push (ParallelOp &&) = 0; )

		virtual uint ProcessMessages () = 0;

		virtual void Quit () = 0;
	};



	class ParallelThread;
	SHARED_POINTER( ParallelThread );



	//
	// Thread Message Queue
	//

	template <typename MTQ>
	struct TThreadMessageQueue
	{
	// types
	private:
		typedef MTQ							mt_queue_t;
		typedef TThreadMessageQueue< MTQ >	Self;

		struct _ProcessMessage
		{
			_ProcessMessage ()
			{}

			void operator () (const ParallelOp &op)
			{
				op.Process();
			}
		};


	// variables
	private:
		mt_queue_t		_messageQueue;
		
		static const usize _MAX_PENDING_QUEUE_COUNT	= 1024;


	// methods
	public:
		TThreadMessageQueue ()
		{
			_messageQueue.ReserveCurrent( 256 );
			_messageQueue.ReservePending( 128 );
		}


		TThreadMessageQueue (const Self &other) : _messageQueue( other._messageQueue )
		{}


		RVALUE_REFERENCE_ONLY(
			TThreadMessageQueue (Self &&other) : _messageQueue( ToRValueRef( other._messageQueue ) )
			{}
		)


		~TThreadMessageQueue ()
		{}


		usize Push (const ParallelOp &op)
		{
			const usize	res = _messageQueue.Push( op );

			if ( res > _MAX_PENDING_QUEUE_COUNT )
			{
				WARNING( "pending queue is too big!" );
				Flush();
				return 0;
			}
			return res;
		}


		RVALUE_REFERENCE_ONLY(
			usize Push (ParallelOp &&op)
			{
				const usize	res = _messageQueue.Push( ToRValueRef( op ) );

				if ( res > _MAX_PENDING_QUEUE_COUNT )
				{
					WARNING( "pending queue is too big!" );
					Flush();
					return 0;
				}
				return res;
			}
		)


		void Flush ()
		{
			return _messageQueue.Flush();
		}


		bool Process ()
		{
			_ProcessMessage	op;
			return _messageQueue.Process( op );
		}


		usize ProcessAll ()
		{
			_ProcessMessage	op;
			return _messageQueue.ProcessAll( op );
		}


		mt_queue_t &	GetQueue ()
		{
			return _messageQueue;
		}
	};


	typedef TThreadMessageQueue<
				TMultiThreadQueue<
					circular_queue< ParallelOp, TSafeOptCopyStrategy<ParallelOp> > > >		ThreadMessageQueue;

	typedef TThreadMessageQueue<
				TMultiThreadQueue2<
					circular_queue< ParallelOp, TSafeOptCopyStrategy<ParallelOp> > > >		ThreadMessageQueue2;



	//
	// Parallel Thread
	//

	class ParallelThread : public IParallelThread
	{
	// variables
	private:
		Thread				_thread;
		ThreadMessageQueue	_messageQueue;
		bool				_quit;


	// methods
	protected:
		static void _ThreadProc (void *parameter)
		{
			ParallelThread *	pthis = (ParallelThread *)parameter;

			pthis->_quit = false;

			while ( not pthis->_quit )
			{
				if ( not pthis->_messageQueue.Process() )
				{
					Thread::Yield();
				}
			}
			pthis->_OnExit();
		}


		void _QuitMessage ()
		{
			_quit = true;
		}


	public:
		ParallelThread () : _quit(false)
		{}


		~ParallelThread ()
		{
			_thread.Wait();
		}


		bool Start (const EngineSubSystemsRef ess)
		{
			_OnStart(ess);
			return _thread.Create( &ParallelThread::_ThreadProc, this );
		}
		

		// IParallelThread //
		bool IsCurrent () const override
		{
			return _thread.IsCurrent();
		}


		void Flush () override
		{
			_messageQueue.Flush();
		}
		
		
		void Push (const ParallelOp &op) override
		{
			_messageQueue.Push( op );
		}


		RVALUE_REFERENCE_ONLY(
			void Push (ParallelOp &&op) override
			{
				_messageQueue.Push( ToRValueRef( op ) );
			}
		)


		usize ProcessMessages () override
		{
			return _messageQueue.ProcessAll();
		}


		void Quit () override
		{
			ParallelOp	op;
			FunctionBuilder::Create( op.func, ParallelThreadPtr(this), &ParallelThread::_QuitMessage );
			Push( ToRValueRef( op ) );

			Flush();

			_thread.Wait();
		}


		static ParallelThreadPtr New ()
		{
			return BaseObject::_New( new ParallelThread() );
		}
	};


}	// Base
}	// Engine